import rx.Observable;
import rx.Subscription;

import java.io.IOException;
import java.util.concurrent.TimeUnit;

/**
 * Created by codeczx on 2017/2/15.
 */
public class ObservableChangeDataTest {

    void testTakeWithIndex(){
        Observable<Integer> values = Observable.create(o->{
            o.onNext(1);
            o.onNext(2);
            o.onError(new Exception("Oops"));
            o.onNext(3);
        });
        Subscription subscription = values
                .take(3)
                .subscribe(System.out::println,
                        e->System.out.println(e.getMessage()),
                        ()->System.out.println("onCompleted")
        );
    }

    void testTakeWithTime(){
        Observable<Long> values = Observable.interval(100, TimeUnit.MILLISECONDS);
        Subscription subscription = values
                .take(500,TimeUnit.MILLISECONDS)
                .subscribe(System.out::println,
                        e->System.out.println(e.getMessage()),
                        ()->System.out.println("onCompleted")
                );
        try {
            System.in.read();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    void testTakeWhile(){
        Observable<Integer> values = Observable.just(1,2,3);
        Subscription subscription = values
                .takeWhile(v->v<2)
                .subscribe(System.out::println);
    }

    void testTakeUntil(){
        Observable<Long> values = Observable.interval(100,TimeUnit.MILLISECONDS);
        Observable<Long> cutOff = Observable.timer(250,TimeUnit.MILLISECONDS);

        Subscription subscription = values
                .takeUntil(cutOff)
                .subscribe(System.out::println,
                e->System.out.println(e.getMessage()),
                ()->System.out.println("onCompleted"));
        try {
            System.in.read();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {
        ObservableChangeDataTest ocdt = new ObservableChangeDataTest();
//        ocdt.testTakeWithIndex();
//        ocdt.testTakeWithTime();
        ocdt.testTakeWhile();
//        ocdt.testTakeUntil();
    }

}
